package drds.binlog.client.adapter.loader;

import drds.binlog.client.Connector;
import drds.binlog.client.Connectors;
import drds.binlog.client.adapter.OuterAdapter;
import drds.binlog.client.impl.ClusterConnector;
import drds.binlog.protocol.Message;

import java.net.SocketAddress;
import java.util.List;
import java.util.concurrent.Executors;

/**
 * 原生canal-server对应的client适配器工作线程
 */
public class AdapterWorker extends AbstractAdapterWorker
{

    private static final int BATCH_SIZE = 50;
    private static final int SO_TIMEOUT = 0;

    private Connector connector;

    /**
     * 单台client适配器worker的构造方法
     *
     * @param destination          canal实例名
     * @param socketAddress        canal-server地址
     * @param outerAdapterListList 外部适配器组
     */
    public AdapterWorker(String destination, SocketAddress socketAddress,
                         List<List<OuterAdapter>> outerAdapterListList)
    {
        this.outerAdapterListList = outerAdapterListList;
        this.destination = destination;
        groupInnerExecutorService = Executors.newFixedThreadPool(outerAdapterListList.size());
        connector = Connectors.newSingleConnector(socketAddress, destination, "", "");
    }

    /**
     * HA模式下client适配器worker的构造方法
     *
     * @param canalDestination   canal实例名
     * @param zookeeperHosts     zookeeper地址
     * @param canalOuterAdapters 外部适配器组
     */
    public AdapterWorker(String canalDestination, String zookeeperHosts,
                         List<List<OuterAdapter>> canalOuterAdapters)
    {
        this.outerAdapterListList = canalOuterAdapters;
        this.destination = canalDestination;
        groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size());
        connector = Connectors.newClusterConnector(zookeeperHosts, canalDestination, "", "");
        ((ClusterConnector) connector).setSoTimeout(SO_TIMEOUT);

        // super.initSwitcher(destination);
    }

    @Override
    public void start()
    {
        if (!running)
        {
            thread = new Thread(new Runnable()
            {

                @Override
                public void run()
                {
                    process();
                }
            });
            thread.setUncaughtExceptionHandler(handler);
            thread.start();
            running = true;
        }
    }

    @Override
    public void stop()
    {
        try
        {
            if (!running)
            {
                return;
            }

            // if (switcher != null && !switcher.state()) {
            // switcher.set(true);
            // }

            connector.stopRunning();
            running = false;

            logger.info("destination {} is waiting for adapters' worker thread die!", destination);
            if (thread != null)
            {
                try
                {
                    thread.join();
                } catch (InterruptedException e)
                {
                    // ignore
                }
            }
            groupInnerExecutorService.shutdown();
            logger.info("destination {} adapters' worker thread dead!", destination);
            for (List<OuterAdapter> outerAdapters : outerAdapterListList)
            {
                for (OuterAdapter adapter : outerAdapters)
                {
                    adapter.destroy();
                }
            }
            logger.info("destination {} all adapters destroyed!", destination);
        } catch (Exception e)
        {
            logger.error(e.getMessage(), e);
        }
    }

    private void process()
    {
        while (!running)
            ; // waiting until running == true
        while (running)
        {
            try
            {
                // if (switcher != null) {
                // switcher.get();
                // }
                logger.info("=============> Start to connect destination: {} <=============", this.destination);
                connector.connect();
                logger.info("=============> Start to subscribe destination: {} <=============", this.destination);
                connector.subscribe();
                logger.info("=============> Subscribe destination: {} succeed <=============", this.destination);
                while (running)
                {
                    // try {
                    // if (switcher != null) {
                    // switcher.get();
                    // }
                    // } catch (TimeoutException e) {
                    // break;
                    // }

                    // server配置canal.instance.network.soTimeout(默认: 30s)
                    // 范围内未与server交互，server将关闭本次socket连接
                    Message message = connector.getWithoutAck(BATCH_SIZE); // 获取指定数量的数据
                    long batchId = message.getId();
                    try
                    {
                        int size = message.getEntryList().size();

                        if (batchId == -1 || size == 0)
                        {
                            try
                            {
                                Thread.sleep(1000);
                            } catch (InterruptedException e)
                            {
                                // ignore
                            }
                        } else
                        {
                            if (logger.isDebugEnabled())
                            {
                                logger.debug("destination: {} batchId: {} batchSize: {} ",
                                        this.destination,
                                        batchId,
                                        size);
                            }
                            long begin = System.currentTimeMillis();
                            writeOut(message);
                            long now = System.currentTimeMillis();
                            if (logger.isDebugEnabled())
                            {
                                logger.debug("destination: {} batchId: {} elapsed time: {} ms",
                                        this.destination,
                                        batchId,
                                        now - begin);
                            }
                        }
                        connector.ack(batchId); // 提交确认
                    } catch (Exception e)
                    {
                        connector.rollback(batchId); // 处理失败, 回滚数据
                        throw e;
                    }
                }

            } catch (Exception e)
            {
                logger.error("process error!", e);
            } finally
            {
                connector.disconnect();
                logger.info("=============> Disconnect destination: {} <=============", this.destination);
            }

            if (running)
            { // is reconnect
                try
                {
                    Thread.sleep(1000);
                } catch (InterruptedException e)
                {
                    // ignore
                }
            }

        }
    }
}
